package com.yl.framework.web.service;

import cn.hutool.json.JSONObject;
import com.google.common.collect.Lists;
import com.yl.common.config.jianke.JianKeProperties;
import com.yl.common.utils.SecurityUtils;
import com.yl.framework.config.WebSocketConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author： yjc
 * @Description： WebSocket操作类
 * ws://192.168.2.230:14000/yl/websocket/jianke?token=VGPRrfdxVYISu0fb3BH0dFNnrxBPtk1ooVe3yCy3KkI=
 * 参数VGPRrfdxVYISu0fb3BH0dFNnrxBPtk1ooVe3yCy3KkI=是建科提供的
 */
@ServerEndpoint(value = "/websocket/jianke",configurator = WebSocketConfig.class)
@Component
@Slf4j
public class WebSocketJiankeSever {

    // 与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    // session集合,存放对应的session
    private static ConcurrentHashMap<String, List<Session>> sessionPool = new ConcurrentHashMap<>();

    /**
     * concurrent包的线程安全Set,用来存放每个客户端对应的WebSocket对象。 key:sessionId val:WebSocketSever
     */
    private static ConcurrentHashMap<String,WebSocketJiankeSever> sessionIdAndWebSocket = new ConcurrentHashMap<>();


    private static JianKeProperties jianKeProperties;

    @Autowired
    public void injectJianKeProperties(JianKeProperties jianKeProperties) {
        //注入jianKeProperties
        setJianKeProperties(jianKeProperties);
    }
    public static void setJianKeProperties(JianKeProperties jianKeProperties){
        WebSocketJiankeSever.jianKeProperties = jianKeProperties;
    }


    /**
     * 解析token 获取userName
     * @param token
     * @return
     */
    private String getUserNameByToken(String token){
        //解密获取用户账号 建科数据解密后 格式如 admin@2024-12-12 00:00:00 账号就是admin
        String userStr = SecurityUtils.decryptAES(token, jianKeProperties.getKey());
        if(com.yl.common.utils.StringUtils.isEmpty(userStr)||!userStr.contains("@")){
            return "";
        }

        String[] split = userStr.split("@");
        String userName = split[0];
        if(com.yl.common.utils.StringUtils.isEmpty(userName)){
            return "";
        }
        return userName;
    }


    /**
     * 建立WebSocket连接
     *
     * @param session
     */
    @OnOpen
    public void onOpen(Session session) {
        // 建立连接
        this.session = session;
        this.addSessionPool(session);
        sessionIdAndWebSocket.put(session.getId(),this);
        log.info("建立连接完成,当前在线人数为：{}", this.getOnUsersNum());
    }

    /**
     * 添加SessionPool
     * @param session 当前session
     */
    private void addSessionPool(Session session){
        String userName = getUserNameByToken(session.getRequestParameterMap().get("token").get(0));
        if(StringUtils.isBlank(userName)){
            try {
                session.close();
            } catch (IOException e) {
                log.error("关闭连接异常->addSessionPool",e);
            }
            return;
        }
        log.info("WebSocket建立连接中,连接用户名：{}",userName);
        List<Session> historySessionList = sessionPool.get(userName);
        if(historySessionList == null){
            historySessionList = Lists.newArrayList();
        }
        historySessionList.add(session);
        sessionPool.put(userName,historySessionList);
    }

    /**
     * 获取在线人数
     * @return
     */
    private int getOnUsersNum(){
        AtomicInteger total = new AtomicInteger();
        sessionPool.forEach((key, val)->{
            if(val.isEmpty()){
                total.addAndGet(1);
            }else{
                total.addAndGet(val.size());
            }
        });
        return total.get();
    }


    /**
     * 发生错误
     *
     * @param throwable e
     */
    @OnError
    public void onError(Throwable throwable,Session session) {
        String userName = getUserNameByToken(session.getRequestParameterMap().get("token").get(0));
        try {
            CloseReason close = new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "关闭客户端，下线！");
            session.close(close);
        } catch (IOException e) {
            log.error("=====onError websocket连接发生错误用户:{}",userName,throwable);
        }
    }

    /**
     * 连接关闭
     */
    @OnClose
    public void onClose(Session session) {
        sessionIdAndWebSocket.remove(session.getId());
        String userName = getUserNameByToken(session.getRequestParameterMap().get("token").get(0));
        this.removeSessionPool(userName,session);
        log.info("连接断开,当前在线人数为：{}", this.getOnUsersNum());
    }

    /**
     * 删除SessionPool
     * @param userName 用户名
     * @param session 当前session
     */
    private void removeSessionPool(String userName, Session session){
        List<Session> historySessionList = sessionPool.get(userName);
        if(historySessionList == null){
            return;
        }
        //过滤后为空直接 删除sessionPool对象的 userId
        historySessionList = historySessionList.stream().filter(e->!e.getId().equals(session.getId())).collect(Collectors.toList());
        if(historySessionList.isEmpty()){
            sessionPool.remove(userName);
            return;
        }
        sessionPool.put(userName,historySessionList);
    }

    /**
     * 接收客户端消息
     *
     * @param message 接收的消息
     */
    @OnMessage
    public void onMessage(String message) {

        String userName = getUserNameByToken(session.getRequestParameterMap().get("token").get(0));

        WebSocketJiankeSever.sendMessageByUser(userName,new JSONObject().set("type","pong").set("time", LocalDateTime.now().toEpochSecond(ZoneOffset.of("+8"))).toString());
        log.info("收到客户端发来的消息：{}", message);
    }



    /**
     * 推送消息到指定用户
     *
     */
    public static void sendMessageByUser(String userName,String message) {
        Assert.notNull(userName,"接收用户名必传");
        log.info("用户名：" + userName + ",推送内容：" + message);
        try {
            List<Session> sessionList = sessionPool.get(userName);
            if(CollectionUtils.isEmpty(sessionList)){
                log.error("===== sendMessageByUser ERROR 用户名:{}不存在",userName);
                return;
            }
            for(Session se : sessionList){
                se.getBasicRemote().sendText(message);
            }
        } catch (IOException e) {
            log.error("推送消息到指定用户发生错误：" + e.getMessage(), e);
        }
    }




    /**
     * 群发消息
     *
     * @param message 发送的消息
     */
    public static void sendAllMessage(String message) {
        log.info("发送消息：{}", message);
        for (Map.Entry<String, WebSocketJiankeSever> entry : sessionIdAndWebSocket.entrySet()) {
            try {
                entry.getValue().session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                log.error("群发消息发生错误：" + e.getMessage(), e);
            }
        }
    }

}

